package com.gitee.xmhzzz.rocketmq.spring.boot.quick.service;

import com.alibaba.fastjson.JSONObject;
import com.gitee.xmhzzz.rocketmq.spring.boot.quick.controller.req.SendMsgReq;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName ExampleService
 * @Description
 * @Author wzq
 * @Date 2024/12/10 14:00
 * @Version 1.0
 */
@Slf4j
@Service
public class ExampleService {

    @Autowired(required = false)
    private RocketMQTemplate rocketMQTemplate;

    private final static String TOPIC = "spring-example-topic";


    /**
      * @Description 异步发送消息
      * @Param [req]
      * @return No such property: code for class: Script1
      */
    public void asyncSendMsg(SendMsgReq req){
        String massage = JSONObject.toJSONString(req.getMsg());
        Message<String> springMsg = MessageBuilder
                .withPayload(massage)
                .setHeader(RocketMQHeaders.KEYS, "key")
                .build();
        rocketMQTemplate.asyncSend(TOPIC +":"+req.getTag(), springMsg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 发送成功逻辑
                log.info("asyncSend ok, messageId[{}]" , sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                // 发送失败逻辑
                log.error("asyncSend failed, payload[{}]", massage);
            }
        },2000);
    }

    /**
      * @Description 同步发送消息
      * @Param [req]
      * @return No such property: code for class: Script1
      */
    public void sendMsg(SendMsgReq req){
        String massage = JSONObject.toJSONString(req.getMsg());
        Message<String> springMsg = MessageBuilder
                .withPayload(massage)
                .setHeader(RocketMQHeaders.KEYS, "key")
                .build();
        rocketMQTemplate.send(TOPIC +":"+req.getTag(),springMsg);
        log.info("send ok, message[{}]" , massage);
    }

    /**
      * @Description 发送单向消息
      * @Param [req]
      * @return No such property: code for class: Script1
      */
    public void sendOneWayMsg(SendMsgReq req){
        String massage = JSONObject.toJSONString(req.getMsg());
        Message<String> springMsg = MessageBuilder
                .withPayload(massage)
                .setHeader(RocketMQHeaders.KEYS, "key")
                .build();
        rocketMQTemplate.sendOneWay(TOPIC +":"+req.getTag(),springMsg);
        log.info("sendOneWay ok, message[{}]" , massage);
    }


    /**
      * @Description 发送顺序消息
      * @Param [req]
      * @return No such property: code for class: Script1
      */
    public void sendOrderlyMsg(SendMsgReq req){
        for (int i = 0; i < 10; i++){
            req.getMsg().put("order",i);
            String massage = JSONObject.toJSONString(req.getMsg());
            Message<String> springMsg = MessageBuilder
                    .withPayload(massage)
                    .setHeader(RocketMQHeaders.KEYS, "key")
                    .build();
            rocketMQTemplate.syncSendOrderly(TOPIC +":"+req.getTag(),springMsg,req.getShardingKey()+i,60000);
            log.info("syncSendOrderly ok, message[{}]" , massage);
        }
    }

    /**
     * @Description 发送延迟消息
     * @Param [req]
     * @return No such property: code for class: Script1
     */
    public void sendDelayedMsg(SendMsgReq req){

        String massage = JSONObject.toJSONString(req.getMsg());
        Message<String> springMsg = MessageBuilder
                .withPayload(massage)
                .setHeader(RocketMQHeaders.KEYS, "key")
                .build();

        log.info("req  message[{}] delayLevel[{}]" , massage,req.getDelayLevel());
        rocketMQTemplate.asyncSend(TOPIC +":"+req.getTag(),springMsg,new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 发送成功逻辑
                log.info("asyncSend ok, messageId[{}]" , sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                // 发送失败逻辑
                log.error("asyncSend failed, payload[{}]", massage);
            }
        },3000,req.getDelayLevel());
        log.info("sendDelayedMsg ok, message[{}]" , massage);
    }


}
